package dim

import java.nio.charset.Charset
import java.util

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.hadoop.hbase.{Cell, HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Result, ResultScanner, Scan, Table}
import org.apache.hadoop.conf
import org.apache.hadoop.hbase.util.Bytes

import scala.collection.mutable

class AreaHBaseReader extends RichSourceFunction[(String,String)]{
  private var connection: Connection = _
  private var table: Table = _
  private var scan: Scan = _
  private val tName = "lagou_area"
  private val family = "f1"

  override def open(parameters: Configuration): Unit = {
    val configuration: conf.Configuration = HBaseConfiguration.create()
    configuration.set("hbase.zookeeper.quorum","linux100")
    configuration.set("hbase.zookeeper.property.clientPort","2181")
    connection = ConnectionFactory.createConnection(configuration)
    table = connection.getTable(TableName.valueOf(tName))
    scan = new Scan()
    scan.addFamily(family.getBytes(Charset.forName("UTF-8")))
  }

  override def run(ctx: SourceFunction.SourceContext[(String,String)]): Unit = {
    val rs: ResultScanner = table.getScanner(scan)
    val iterator: util.Iterator[Result] = rs.iterator()
    while (iterator.hasNext) {
      val result: Result = iterator.next()
      val rowKey: String = Bytes.toString(result.getRow)
      val cells: util.List[Cell] = result.listCells()
      val stringBuilder = new StringBuilder
      cells.forEach((cell: Cell) => {
        stringBuilder.append(Bytes.toString(cell.getValueArray, cell.getValueOffset, cell.getValueLength)).append("-")
      })
      val str: String = stringBuilder.deleteCharAt(stringBuilder.length - 1).toString()
      ctx.collect((rowKey, str))
    }
  }

  override def cancel(): Unit = {
    try {
      if(table != null) {
        table.close()
      }
      if(connection != null) {
        connection.close()
      }
    } catch {
      case e:Exception => println(e.getMessage)
    }
  }


}
